#include "channel.h"
#include <unistd.h>

namespace ProcessRouter {

static void listener_child_cb (EV_P_ struct ev_child *w, int revents)
{
    printf("callback %s %d %d %p\n", __FUNCTION__, w->pid, getpid(), loop);

    std::list<Listener*> *listeners = (std::list<Listener*>*)ev_userdata(loop);
    std::list<Listener*>::iterator it = listeners->begin();
    for (; it != listeners->end(); it++) {
        if (w->rpid == ((Channel*)(*it))->childPid()) {
            if ((*it)) {
                (*it)->onError();
            }
            break;
        }
    }
    //printf ("process %d exited with status %x\n", w->rpid, w->rstatus); 
}

CMessageEvent::CMessageEvent(CMessage msg, Type type)
    : Rooftop::EventBase(MESSAGEEVENT)
    , m_message(msg)
    , m_type(type)
{
}

Channel::Channel(IOThread *thread)
    : Listener(thread)
{
}

Channel::~Channel()
{
}

void Channel::setClient(ChannelHandle *client)
{
    client->setChannel(this);
    Listener::setClient(client);
}

ChannelHandle::ChannelHandle()
    : m_channel(NULL)
{
}

void ChannelHandle::setChannel(Channel *channel)
{
    m_channel = channel;
}

Channel* ChannelHandle::channel()
{
    return m_channel;
}

FifoChannel::FifoChannel(IOThread *thread)
    : Channel(thread)
    , m_isChild(false)
    , m_pid(getpid())
{
}

FifoChannel::~FifoChannel()
{
}

void FifoChannel::open()
{
    int ret = 0;

    ret = pipe(m_pfds);
    if (ret == -1)
        perror("Create pipe error");

    ret = pipe(m_cfds);
    if (ret == -1)
        perror("Create pipe error");
}

void FifoChannel::destroy()
{
    ev_child_stop (currentThread()->loopStruct(), &m_cw); 
    Listener::remove();
    close(readFd());
    close(writeFd());
}

CMessage FifoChannel::recv()
{
    int ret = 0;
    std::string content;

    int size = 0;
    ret = read(readFd(),(char*)&size, sizeof(int));

    if (ret <= 0)
        return CMessage();

    ret = 0;

    char buf[1024];

    int tsize = size;
    do {
        int nsize = 0;
        if (tsize <= 1024)
            nsize = tsize;
        else
            nsize = 1024;

        memset(buf, 0, 1024);

        int tret = read(readFd(), buf, nsize);
        if (tret == 0) {
            sleep(10);
            continue;
        }

        if (tret < 0)
            break;

        tsize -= tret;
        ret += tret;

        content.append(buf, tret);
    } while(tsize > 0);

    if (ret <= 0)
        return CMessage();

    if (ret != size)
        std::cout << "read error " << ret << " | " <<  size << std::endl;

    return CMessage(content);
}

int FifoChannel::send(CMessage &message)
{
    if (message.isNull())
        return -1;
    
    std::string content = message.content();
    int size = message.content().size();
    content.insert(0, (char*)&size, sizeof(int));

    int ret = write(writeFd(), content.data(), content.size());
    if (ret == -1) {
        perror("Fifo send error");
    }

    return ret;
}

void FifoChannel::forkInit(ChannelHandle *client)
{
    if (m_isChild) {
        close(m_cfds[1]);
        close(m_pfds[0]);
    } else {
        close(m_cfds[0]);
        close(m_pfds[1]);
    }

    init(readFd());

    setClient(client);
}

int FifoChannel::readFd()
{
    if (m_isChild)
        return m_cfds[0];
    else
        return m_pfds[0];
}

int FifoChannel::writeFd()
{
    if (m_isChild)
        return m_pfds[1];
    else
        return m_cfds[1];
}

void FifoChannel::setChild()
{
    m_isChild = true;
}

int FifoChannel::cfork()
{
    int pid = fork();
    if (pid == 0) {
        setChild();
        m_pid = getpid();
        currentThread()->forkLoop();
    } else if (pid > 0) {
        setChildPid(pid);
        ev_child_init (&m_cw, listener_child_cb, pid, 0); 
        ev_set_priority(&m_cw, 1);
        ev_child_start (currentThread()->loopStruct(), &m_cw); 
    }
    return pid;
}

void ZAddress::toString()
{
    byte *data = m_addr;
    size_t size = m_size;

    std::string retstring;
    uint char_nbr;

    std::stringstream ss;
    for (char_nbr = 0; char_nbr < size; char_nbr++) {
        int num = data [char_nbr];
        ss << num;
    }
    m_str = ss.str();
}

std::string CMessage::addressString()
{
    if (!m_address.get())
        return "";

    return m_address->str();
}

byte* CMessage::address()
{
    if (!m_address.get())
        return NULL;

    return m_address->m_addr;
}

void CMessage::setAddr(byte* addr, size_t size)
{
    boost::shared_ptr<ZAddress> address(new ZAddress(addr, size));
    m_address = address;
}

ZmqChannel::ZmqChannel(IOThread *thread)
    : Channel(thread)
    , m_zmqctx(NULL)
    , m_zmqsocket(NULL)
{
}

ZmqChannel::~ZmqChannel()
{
}

void ZmqChannel::setProtocol(std::string protocol)
{
    m_protocol = protocol;
}

std::string ZmqChannel::protocol()
{
    return m_protocol;
}

void ZmqChannel::open()
{
    m_zmqctx = zctx_new();
    m_zmqsocket = zsocket_new(m_zmqctx, ZMQ_ROUTER);

    int socketfd;
    size_t fdlen = sizeof(socketfd);
    zmq_getsockopt(m_zmqsocket, ZMQ_FD, &socketfd, &fdlen);

    if (m_protocol == "") {
        printf("Protocol is null!\n");
        return;
    }

    int ret = 0;
    ret = zsocket_bind(m_zmqsocket, m_protocol.c_str());
    if (ret == -1) {
        perror("zsocket bind error");
        return;
    }

    init(socketfd);
}

void ZmqChannel::destroy()
{
    currentThread()->removeListener((Listener*)this);
    if (m_zmqsocket)
        zmq_close(m_zmqsocket);
    if (m_zmqctx)
        zmq_term(m_zmqctx);
    Listener::remove();
}

CMessage ZmqChannel::recv()
{
    std::string message;
    zmsg_t *msg = zmsg_recv_nowait(m_zmqsocket);
    if (!msg) {
        return CMessage();
    }

    message = getMessage(msg);
    byte* address = NULL;
    size_t addrsize = 0;
    getAddress(msg, &address, &addrsize);  

    CMessage zmsg(message);
    zmsg.setAddr(address, addrsize);

    zmsg_destroy(&msg);
    return zmsg;
}

int ZmqChannel::send(CMessage &message)
{
    CMessage *zm = (CMessage*)&message;
    zmsg_t *msg = zmsg_new();
    byte* address = zm->address();
    zmsg_addmem(msg, address, 5);
    zmsg_addmem(msg, "", 0);
    zmsg_addmem(msg, zm->content().c_str(), zm->content().size());
    zmsg_send(&msg, m_zmqsocket);
    zmsg_destroy(&msg);
}

//private
std::string ZmqChannel::getMessage(zmsg_t* msg)
{
    zframe_t *frame = zmsg_last(msg);
    byte *data = zframe_data(frame);
    size_t size = zframe_size(frame);
    std::string message((char*)data, size);
    return message;
}

void ZmqChannel::getAddress(zmsg_t* msg, byte** _addr, size_t *_size)
{
    zframe_t* addr = zmsg_first(msg);
    //zframe_print(addr, "Address:\n");
    *_addr = zframe_data(addr);
    *_size = zframe_size(addr);
}

}


